
package be.ibridge.kettle.UpdateRRD;

import java.io.IOException;

import org.jrobin.core.RrdDb;
import org.jrobin.core.RrdException;
import org.jrobin.core.Sample;
import org.pentaho.di.core.Const;
import org.pentaho.di.core.exception.KettleException;
import org.pentaho.di.core.row.RowMetaInterface;
import org.pentaho.di.trans.Trans;
import org.pentaho.di.trans.TransMeta;
import org.pentaho.di.trans.step.BaseStep;
import org.pentaho.di.trans.step.StepDataInterface;
import org.pentaho.di.trans.step.StepInterface;
import org.pentaho.di.trans.step.StepMeta;
import org.pentaho.di.trans.step.StepMetaInterface;




/*
 * Created on 2-jun-2003
 *
 */

public class UpdateRRDPlugin extends BaseStep implements StepInterface
{
    private UpdateRRDPluginData data;
	private UpdateRRDPluginMeta meta;
	
	public UpdateRRDPlugin(StepMeta s, StepDataInterface stepDataInterface, int c, TransMeta t, Trans dis)
	{
		super(s,stepDataInterface,c,t,dis);
	}
	
	public boolean processRow(StepMetaInterface smi, StepDataInterface sdi) throws KettleException
	{
	    meta = (UpdateRRDPluginMeta)smi;
	    data = (UpdateRRDPluginData)sdi;
	    
		Object[] r=getRow();    // get row, blocks when needed!
		if (r==null)  // no more input to be expected...
		{
			setOutputDone();
			return false;
		}
        
        if (first)
        {
            first = false;
            data.outputRowMeta = (RowMetaInterface)getInputRowMeta().clone();
            meta.getFields(data.outputRowMeta, getStepname(), null, null, this);      
        }

        String RRDFilenm = meta.getRRDFilename();
        //logBasic("Output : " + RRDFilenm);
        //logBasic("L1Col1 : " + r[0].toString());
        //logBasic("L1Col2 : " + r[1].toString());
        	
        try {
        	
			UpdateRRDFile(RRDFilenm,r[0].toString(),r[1].toString());
			logBasic("RRD File " + RRDFilenm + " updated with values : " + r[0].toString() + ":" + r[1].toString() );
		} catch (RrdException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		} catch (IOException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
        
        
		if (checkFeedback(linesRead)) logBasic("Linenr "+linesRead);  // Some basic logging every 5000 rows.
			
		return true;
	}
	
	public static void UpdateRRDFile(String RRDFile, String Timest, String Val) throws RrdException, IOException{
		RrdDb rrdDb = new RrdDb(RRDFile);
		Sample sample = rrdDb.createSample();
		sample.setAndUpdate(Timest +":" + Val);
		rrdDb.close();
	}
	
	
		
	public boolean init(StepMetaInterface smi, StepDataInterface sdi)
	{
	    meta = (UpdateRRDPluginMeta)smi;
	    data = (UpdateRRDPluginData)sdi;

	    return super.init(smi, sdi);
	}

	public void dispose(StepMetaInterface smi, StepDataInterface sdi)
	{
	    meta = (UpdateRRDPluginMeta)smi;
	    data = (UpdateRRDPluginData)sdi;

	    super.dispose(smi, sdi);
	}
	
	//
	// Run is were the action happens!
	public void run()
	{
		logBasic("Starting to run...");
		try
		{
			while (processRow(meta, data) && !isStopped());
		}
		catch(Exception e)
		{
			logError("Unexpected error : "+e.toString());
            logError(Const.getStackTracker(e));
			setErrors(1);
			stopAll();
		}
		finally
		{
		    dispose(meta, data);
			logBasic("Finished, processing "+linesRead+" rows");
			markStop();
		}
	}
}
